package com.yanggu.bigdata.realtime.app.dws

import cn.hutool.core.date.DateUtil
import cn.hutool.core.util.StrUtil
import com.alibaba.fastjson.JSON
import com.yanggu.bigdata.realtime.bean.VisitorStats
import com.yanggu.bigdata.realtime.common.GmallConfig.KAFKA_BROKER_LIST
import com.yanggu.bigdata.realtime.utils.{ClickHouseUtil, KafkaUtil}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.time.Duration
import java.util.Date

/**
 * 访客主题
 */
object VisitorStatsApp {

  def main(args: Array[String]): Unit = {
    //1. 创建执行环境
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    //2. Kafka 的 pv、uv、跳转明细主题中获取数据, 并且转换数据格式
    val groupId: String = "visitor_stats_app"
    val pageViewSourceTopic = "dwd_page_log"
    val uniqueVisitSourceTopic = "dwm_unique_visit"
    val userJumpDetailSourceTopic = "dwm_user_jump_detail"

    val pageViewKafkaDataStream = KafkaUtil.getKafkaDataStream(environment, KAFKA_BROKER_LIST, pageViewSourceTopic, groupId)

    //pv数据流
    val pageViewDataStream = pageViewKafkaDataStream
      .map(json => {
        val data = JSON.parseObject(json)
        val common = data.getJSONObject("common")
        new VisitorStats("", "",
          common.getString("vc"),
          common.getString("ch"),
          common.getString("ar"),
          common.getString("is_new"),
          0L, 1L, 0L, 0L, data.getJSONObject("page").getLong("during_time"), data.getLong("ts"))
      })

    //uv数据流
    val uniqueVisitDataStream = KafkaUtil.getKafkaDataStream(environment, KAFKA_BROKER_LIST, uniqueVisitSourceTopic, groupId)
      .map(json => {
        val jsonObj = JSON.parseObject(json)
        val common = jsonObj.getJSONObject("common")
        new VisitorStats("", "",
          common.getString("vc"),
          common.getString("ch"),
          common.getString("ar"),
          common.getString("is_new"),
          1L, 0L, 0L, 0L, 0L, jsonObj.getLong("ts"));
      })

    //sessionVisit数据流
    val sessionVisitDataStream = pageViewKafkaDataStream.flatMap(new FlatMapFunction[String, VisitorStats] {
      override def flatMap(json: String, out: Collector[VisitorStats]): Unit = {
        val jsonObj = JSON.parseObject(json)
        val lastPageId = jsonObj.getJSONObject("page").getString("last_page_id")
        if (StrUtil.isEmpty(lastPageId)) {
          val common = jsonObj.getJSONObject("common")
          out.collect(new VisitorStats("", "",
            common.getString("vc"),
            common.getString("ch"),
            common.getString("ar"),
            common.getString("is_new"),
            0L, 0L, 1L, 0L, 0L, jsonObj.getLong("ts")))
        }
      }
    })

    //uj数据流
    val userJumpDetailDataStream = KafkaUtil.getKafkaDataStream(environment, KAFKA_BROKER_LIST, userJumpDetailSourceTopic, groupId)
      .map(json => {
        val jsonObj = JSON.parseObject(json)
        val common = jsonObj.getJSONObject("common")
        new VisitorStats("", "",
          common.getString("vc"),
          common.getString("ch"),
          common.getString("ar"),
          common.getString("is_new"),
          0L, 0L, 0L, 1L, 0L, jsonObj.getLong("ts"))
        }
      )

    //3. union所有的流
    val resultDataStream = pageViewDataStream.union(uniqueVisitDataStream, sessionVisitDataStream, userJumpDetailDataStream)
      //4. 设置watermark和延迟
      .assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(2L))
        .withTimestampAssigner(new SerializableTimestampAssigner[VisitorStats] {
          override def extractTimestamp(element: VisitorStats, recordTimestamp: Long): Long = element.getTs
        }))
      .keyBy(data => (data.getVc, data.getCh, data.getAr, data.getIs_new))
      .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
      .allowedLateness(Time.seconds(2L))
      //5. 开窗聚合统计
      .reduce((value1, value2) => {
        //pv相加
        value1.setPv_ct(value1.getPv_ct + value2.getPv_ct)
        //uv相加
        value1.setUv_ct(value1.getUv_ct + value2.getUv_ct)
        //uj相加
        value1.setUj_ct(value1.getUj_ct + value2.getUj_ct)
        //sv相加
        value1.setSv_ct(value1.getSv_ct + value2.getSv_ct)
        //dur相加
        value1.setDur_sum(value1.getDur_sum + value2.getDur_sum)
        value1
      }, new ProcessWindowFunction[VisitorStats, VisitorStats, (String, String, String, String), TimeWindow] {
        override def process(key: (String, String, String, String), context: Context, elements: Iterable[VisitorStats], out: Collector[VisitorStats]): Unit = {
          val data = elements.head
          val window = context.window
          //补充窗口的开始和结束时间字段
          data.setStt(DateUtil.formatDateTime(new Date(window.getStart)))
          data.setEdt(DateUtil.formatDateTime(new Date(window.getEnd)))
          out.collect(data)
        }
      })
    resultDataStream.print("result>>>>>>>>")

    //6. 写入到clickhouse中
    resultDataStream
      .addSink(ClickHouseUtil.getSinkFunction[VisitorStats]("INSERT INTO visitor_stats VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"))

    //7. 启动Job
    environment.execute("VisitorStatsApp Job")

  }

}
